package com.atguigu.interview.study.thread;

import java.util.concurrent.*;

/**
 * @author gcq
 * @Create 2021-01-02
 */
public class CompletableFutureDemo {
    public static ExecutorService executor = Executors.newFixedThreadPool(10);

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        /**
         * thenApplyAsync; 能接受上一步返回结果,有返回值
         */
//        getCompletableFutureSupplyAsyncAndThenApplyAsync();

        /**
         * thenAcceptAsync能接受上一步结果，但是没有返回值
         */
//        getCompletableFutureSuppluAsyncThenAcceptAsync();

        /**
         * thenRun不能获取到上一步的执行效果 无返回值
         */
//        getCompletableFutureSupplyAsyncThenRunAsync();

        /**
         * runAfterBoth:组合两个 future，不需要获取 future 的结果，只需要两个 future处理完成任务后，处理该任务，
         */
//        getCompletableFutureRunAfterBothAsync();

        /**
         * thenAccpetBoth: 组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值
         */
//        getCompletableFutureThenAcceptBothAsync();

        /**
         * thenCombine: 组合两个 future，获取两个 future的返回结果，并返回当前任务的返回值
         */
//        getCompletableFutureThenCombineAsync();

        /**
         * 为了能获取任务线程内发生的异常，需要使用CompletableFuture的CompleExceptionally方法将导致CompletableFuture内发生的问题异常抛出
         */
//        getCompletableFutureCompleteExceptionally();

        /**
         *  anyOf 只要有一个任务完成就行
         */
//        getCompletableAnyOf();

        /**
         * allof 等待所有任务都完成
         */
//        getCompletableFutureAllOf();

        /**
         * 可以对结果做最后的处理（可处理异常），可改变返回值
         */
//        getCompletableFutureHandle();
    }

    /**
     * 可以对结果做最后的处理（可处理异常），可改变返回值
     */
    private static void getCompletableFutureHandle() {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).handle((res,thr) ->{
            // 可以对结果做最后的处理（可处理异常），可改变返回值
            if (res != null ) {
                System.out.println(res);
                return res * 2;
            }
            if (thr != null) {
                return 0;
            }
            return 0;
        });
    }

    /**
     * allof 等待所有任务都完成
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void getCompletableFutureAllOf() throws InterruptedException, ExecutionException {
        CompletableFuture<String> futureImg = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的图片信息");
            return "hello.jpg";
        }, executor);

        CompletableFuture<String> futureAttr = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的属性");
            return "黑色+156GB";
        }, executor);

        CompletableFuture<String> futureDesc = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
                System.out.println("查询商品介绍");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }  return "华为";
        }, executor);

        CompletableFuture<Void> allOf = CompletableFuture.allOf(futureImg, futureAttr, futureDesc);
        // 等待所有任务都完成了
        allOf.get();
        System.out.println("main....end" + allOf.get());
    }

    /**
     * anyOf 只要有一个任务完成就行
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void getCompletableAnyOf() throws InterruptedException, ExecutionException {
        CompletableFuture<String> futureImg = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的图片信息");
            return "hello.jpg";
        }, executor);

        CompletableFuture<String> futureAttr = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的属性");
            return "黑色+156GB";
        }, executor);

        CompletableFuture<String> futureDesc = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
                System.out.println("查询商品介绍");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }  return "华为";
        }, executor);

        CompletableFuture<Object> allOf = CompletableFuture.anyOf(futureImg, futureAttr, futureDesc);
        // 只要有一个任务完成就行
        allOf.get();
        System.out.println("main....end" + allOf.get());
    }

    /**
     * 为了能获取任务线程内发生的异常，需要使用CompletableFuture的CompleExceptionally方法将导致CompletableFuture内发生的问题异常抛出
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void getCompletableFutureCompleteExceptionally() throws InterruptedException, ExecutionException {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        new Thread(() -> {
            // 模拟执行耗时任务
            System.out.println("task doing...");
            try {
                Thread.sleep(3000);
                int i = 10/5;
            } catch (Exception e) {
                // 告诉completableFuture任务发生异常了
                completableFuture.completeExceptionally(e);
            }
            // 告诉completableFuture任务已经完成
            completableFuture.complete("ok");
        }).start();
        // 获取任务结果，如果没有完成会一直阻塞等待
        String result = completableFuture.get();
        System.out.println("计算结果:" + result);
    }

    /**
     * thenCombine: 组合两个 future，获取两个 future的返回结果，并返回当前任务的返回值
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void getCompletableFutureThenCombineAsync() throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1线程:" + Thread.currentThread().getId());
            int i = 10 / 4;
            System.out.println("任务1运行结果:" + i);
            return i;
        }, executor);

        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务2线程:" + Thread.currentThread().getId());
            int i = 10 / 4;
            System.out.println("任务2运行结果:");
            return "Hello";
        }, executor);
        //      返回f1和f2的结果
        // thenCombine: 组合两个 future，获取两个 future的返回结果，并返回当前任务的返回值
        CompletableFuture<String> future03 = future01.thenCombineAsync(future02, (f1, f2) -> {
            return f1 + ":" + f2 + "-> HaHa";
        }, executor);
        System.out.println(future03.get());
    }

    /**
     *  thenAccpetBoth: 组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值
     */
    private static void getCompletableFutureThenAcceptBothAsync() {
        CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1线程:" + Thread.currentThread().getId());
            int i = 10 / 4;
            System.out.println("任务1运行结果:" + i);
            return i;
        }, executor);

        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务2线程:" + Thread.currentThread().getId());
            int i = 10 / 4;
            System.out.println("任务2运行结果:");
            return "Hello";
        }, executor);

        // thenAccpetBoth: 组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值
        future01.thenAcceptBothAsync(future02,(f1,f2)->{
            System.out.println("任务三开始。。。之前结果为:" + f1 +"-->" + f2);
        },executor);
    }

    /**
     * runAfterBoth:组合两个 future，不需要获取 future 的结果，只需要两个 future处理完成任务后，处理该任务，
     */
    private static void getCompletableFutureRunAfterBothAsync() {
        CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1线程:" + Thread.currentThread().getId());
            int i = 10 / 4;
            System.out.println("任务1运行结果:" + i);
            return i;
        }, executor);

        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务2线程:" + Thread.currentThread().getId());
            int i = 10 / 4;
            System.out.println("任务2运行结果:");
            return "Hello";
        }, executor);

        //runAfterBoth:组合两个 future，不需要获取 future 的结果，只需要两个 future处理完成任务后，处理该任务，
        future01.runAfterBothAsync(future02,()->{
            System.out.println("任务3开始...");
        },executor);
    }

    /**
     * thenRun不能获取到上一步的执行效果 无返回值
     */
    private static void getCompletableFutureSupplyAsyncThenRunAsync() {
        CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("运行结果:" + i);
            return i;
        }).thenRunAsync(() -> {
            // 不能获取到上一步的执行效果 无返回值
            System.out.println("居然没有返回值嘿嘿");
        });
    }

    /**
     *  thenAcceptAsync能接受上一步结果，但是没有返回值
     */
    private static void getCompletableFutureSuppluAsyncThenAcceptAsync() {
        CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("运行结果:" + i);
            return i;
        }).thenAcceptAsync(res -> {
            // 能接收到上一步结果，但是无法返回值
            System.out.println("返回结果是：" + res);
        });
    }

    /**
     * thenApplyAsync; 能接受上一步返回结果,有返回值
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void getCompletableFutureSupplyAsyncAndThenApplyAsync() throws InterruptedException, ExecutionException {
        CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("运行结果:" + i);
            return i;
        }).thenApplyAsync(res -> {
            // 接收上一步结果，并且能更改，能更改成自己想要的返回值
            System.out.println("返回的结果是：" + res);
            return "hello" + res;
        });
        System.out.println(stringCompletableFuture.get());
    }

    /**
     * whenComplete 计算完成时的回调方法
     * exceptionally 感知异常并返回默认值
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void getCompletableFutureSupplyAsyncAndWhenComplete() throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10 / 0;
            System.out.println("运行结果:" + i);
            return i;
        }).whenComplete((res, exception) -> { // res 返回结果 exception 异常
            // 能感知到异常信息和结果，但是无法修改返回的数据
            System.out.println("异步任务执行完成。。。结果是:" + res + " 异常是:" + exception);
        }).exceptionally(throwable -> {
            // 可以感知异常，并且返回默认值
            return 10;
        });
        System.out.println(integerCompletableFuture.get());
    }

    /**
     * Supplier供给型接口 有返回值
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void getCompletableFutureSupplyAsyncMethodParamsSupplier() throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10 / 5;
            System.out.println("运行结果:" + i);
            return i;
        });
        // 通过get拿到返回结果
        System.out.println(integerCompletableFuture.get());
    }

    /**
     *  以Runnable接口作为参数的，没有返回结果
     */
    private static void getCompletableFutureRunAsyncMethodParamsRunnable() {
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果:" + i);
        });
        System.out.println(completableFuture);
    }

    /**
     * ThreadPoolExecutor 线程池复习
     * @param args
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void thread(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main...start");
        /**
         * 1)、继承Thread
         *        Thread01 thread01 = new Thread01();
         *         thread01.start();
         * 2)、实现Runnable接口
         *         Runnable01 runnable01 = new Runnable01();
         *         new Thread(runnable01).start();
         * 3)、实现Callable接口 + FutureTask (可以拿到返回结果，可以处理异常)
         *       FutureTask<Integer> integerFutureTask = new FutureTask<>(new Callable01());
         *         new Thread(integerFutureTask).start();
         *
         *         //阻塞等待线程执行完成 获取返回结果
         *         Integer integer = integerFutureTask.get();
         *         System.out.println(integer);
         * 4)、线程池
         *        给线程池直接提交任务
         */

        /**
         * corePoolSize 保留在池中的线程数 即使处于空闲状态 除非设置了allowCoreThreadTimeOut
         *
         * maximumPoolSize *池中允许的最大线程数
         *
         * keepalivueTime 存活时间 如果当前线程大于core的数量
         *          释放空闲的线程 maximumPoolsize-corePoolSize 只要线程空闲大于指定的keepAlivuetime
         * unit:时间单位
         * BlockingQUeue<Runnable> workQueue 阻塞队列 如果任务有很多 就会将目前多的任务放在队列里面
         *      只要有线程空闲，就会去队列里面取出新的任务继续执行
         * threadFactory 线程创建工厂
         * RejectedExecutionHandler 如果队列满了 按照我们指定得拒绝策略拒绝指定任务
         *
         * 工作顺序
         * 1)、线程池创建好 准备好core数量的核心线程，准备接受任务
         * 1.1、core满了 就将在进来的任务放入阻塞队列中 空闲的core就会自己去阻塞队列获取任务执行
         * 1.2、阻塞队列满了 就直接开新线程执行 最大只能开到max指定数量
         * 1.3、max满了就用RejectedExecutionHandler 拒绝任务
         * 1.4、max都执行完成，有很多空闲 指定时间以后keepAlivueTime以后 释放max-core这些线程
         *
         *       new LinkedBlockingQueue<>() 默认是Integer最大值 内存不够
         *
         *       一个线程池 core:7 max:20 queue:50 100并发进来怎么分配
         *       7个会立即得到执行 50个进入队列 再开13个进行执行，剩下的30个就使用拒绝策略
         *       如果不想抛弃还要执行 CallerRunsPolicy
         */
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5,
                200,
                10,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
//          Executors.newCachedThreadPool() core是0 所有都可回收
//          Executors.newFixedThreadPool() 固定大小 core=max 都不可以回收
//          Executors.newScheduledThreadPool() 定时任务的线程池
//            Executors.newSingleThreadExecutor() 单线程的线程池,后台从队列里面获取任务 挨个执行
        System.out.println("main....end");
    }

    /**
     * 继承Thread类
     */
    public static class Thread01 extends Thread{
        @Override
        public void run() {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10/2;
            System.out.println("运行结果:" + i);
        }
    }

    /**
     * 实现Runnable接口
     */
    public static class Runnable01 implements Runnable{
        @Override
        public void run() {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10/2;
            System.out.println("运行结果:" + i);
        }
    }

    /**
     * 实现Callable接口
     */
    public static class Callable01 implements Callable<Integer> {
        @Override
        public Integer call() throws Exception {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10/2;
            System.out.println("运行结果:" + i);
            return i;
        }
    }
}